// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/base/executor/archive_executor.h"

#include "coin2/exchange/base/executor/archive_system.h"
#include "coin2/exchange/base/feed/system.h"

namespace coin2::exchange::base::executor {

using ::coin2::exchange::base::executor::ArchiveMixin;

namespace impl {

void ArchiveSyncExecutor::Run() {
  Prepare();
  // TODO(donggu): change interface
  CHECK_GE_THROW(subsystems_.size(), 1) << "at least one archive subsystem is required";

  std::vector<ArchiveMixin*> systems;
  for (auto& subsystem : subsystems_) {
    systems.push_back(dynamic_cast<ArchiveMixin*>(subsystem));
  }

  if (config_.merge_sort()) {
    std::vector<Entry> entries;
    // init
    {
      for (size_t i = 0; i < systems.size(); i++) {
        if (auto ts = systems[i]->PeekTimestamp()) {
          entries.push_back({i, ts});
        }
      }
    }
    // run
    while (!entries.empty()) {
      size_t e = 0;
      for (size_t i = 1; i < entries.size(); i++) {
        if (entries[i] < entries[e]) {
          e = i;
        }
      }

      systems[entries[e].index]->ConsumeRecord();

      // pull next record
      if (auto ts = systems[entries[e].index]->PeekTimestamp()) {
        // update timestamp
        entries[e].timestamp = ts;
      } else {  // EOF?
        // pop
        entries[e] = entries.back();  // bring the last
        entries.pop_back();
      }
    }
  } else {
    // no merge
    // useful for feed check, plotting, etc
    for (auto& system : systems) {
      while (system->PeekTimestamp() != 0) {
        system->ConsumeRecord();
      }
    }
  }

  Cleanup();
}

}  // namespace impl
}  // namespace coin2::exchange::base::executor
